package hotspotkey

import (
	"context"
	"github.com/cespare/xxhash/v2"
	"github.com/coocood/freecache"
	"go.uber.org/zap"
	"sync"
	"time"
)

// TODO  threshold 应该能动态调整。

// HotSpotCallBack 触发热点回调
// shortId 短链接id
// value 需要缓存的数据
// err 发生错误
type HotSpotCallBack func(ctx context.Context, shortId string) (value string, err error)

// HotSpotDetector 热点探测器
type HotSpotDetector struct {
	shardMask uint64          // 快速取模掩码
	shards    []*shard        // 分片
	callback  HotSpotCallBack // 触发热点回调
	cache     *freecache.Cache
	logger    *zap.Logger
}

type shard struct {
	sync.RWMutex

	sws map[string]*slidingWindow
}

type slidingWindow struct {
	sync.Mutex

	windowSize time.Duration // 时间窗口 例如 5分钟
	threshold  int           // 阈值
	timestamps []time.Time   // 时间戳环形队列
	start      int           // 有效数据起始索引
	count      int           // 有效数据量
}

func newSlidingWindow(windowSize time.Duration, threshold int) *slidingWindow {
	return &slidingWindow{
		windowSize: windowSize,
		threshold:  threshold,
		timestamps: make([]time.Time, threshold*2),
	}
}

func NewHotSpotDetector(logger *zap.Logger) *HotSpotDetector {
	shardCount := 256
	shards := make([]*shard, shardCount)

	for i := range shards {
		shards[i] = &shard{
			sws: make(map[string]*slidingWindow),
		}
	}

	return &HotSpotDetector{
		logger:    logger,
		shards:    shards,
		shardMask: uint64(shardCount) - 1,
		cache:     freecache.NewCache(100 * 1024 * 1024),
	}
}

func (d *HotSpotDetector) RegisterCallback(callback HotSpotCallBack) {
	d.callback = callback
}

// GetHotspot 获取热点数据
func (d *HotSpotDetector) GetHotspot(ctx context.Context, shortId string) (string, bool) {
	if value, err := d.cache.Get([]byte(shortId)); err == nil {
		if value != nil {
			return string(value), true
		}
	}

	shard := d.getShard(shortId)

	shard.RLock()
	sw, ok := shard.sws[shortId]
	shard.RUnlock()

	if !ok {
		shard.Lock()
		sw = newSlidingWindow(5*time.Minute, 5000)
		shard.sws[shortId] = sw
		shard.Unlock()
	}

	if sw.incr() {
		v, err := d.callback(ctx, shortId)
		if err != nil {
			d.logger.Error("触发热点回调失败", zap.String("shortId", shortId), zap.Error(err))
			return "", false // 降级为非热点数据
		}

		if v != "" {
			err := d.cache.Set([]byte(shortId), []byte(v), 300)
			if err != nil {
				d.logger.Error("触发热点回调本地缓存失败", zap.String("shortId", shortId), zap.Error(err))
				return "", false
			}

			return v, true
		}

		return "", false
	}

	return "", false
}

// getShard 获取分片
func (d *HotSpotDetector) getShard(shorId string) *shard {
	key := xxhash.Sum64String(shorId)
	return d.shards[key&d.shardMask]
}

// incr 累计计数
func (sw *slidingWindow) incr() bool {
	sw.Lock()
	defer sw.Unlock()

	now := time.Now()

	sw.evict(now) // 清理过期数据

	pos := (sw.start + sw.count) % len(sw.timestamps)
	sw.timestamps[pos] = now

	if sw.count < len(sw.timestamps) {
		sw.count++
	} else {
		sw.start = (sw.start + 1) % len(sw.timestamps)
	}
	return sw.count >= sw.threshold
}

// evict 清理滑动窗口
func (sw *slidingWindow) evict(now time.Time) {
	// 计算窗口开始时间
	// now = 12:07
	// windowStartTime = 12:07 - 5分钟  = 12:02
	// 窗口开始时间为 12:02, 在这之前的都过期咯
	windowStartTime := now.Add(-sw.windowSize)

	for sw.count > 0 {
		// 有效数据的起始位置时间在新的窗口开始时间之后；则是有效数据；
		if sw.timestamps[sw.start].After(windowStartTime) {
			break
		}

		// start 向后移动一个元素
		sw.start = (sw.start + 1) % len(sw.timestamps)
		sw.count--
	}
}
